package thread.example.consume01;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 由以下结论得出,多线程比单线程快无数倍
 * @author jiangwz
 * @create 2019/9/29.
 */
public class SingleConsumer {

    public static LinkedBlockingQueue<Integer> msgQueue = new LinkedBlockingQueue<>(10000);

    public void init(int num){
        for (int i = 0; i < num; i++) {
            msgQueue.offer(i);
        }
    }

    /**
     * 直接运行该方法有问题,无限循环,加了volatile也没用
     */
    public void run() {
        Integer poll = msgQueue.poll();
        System.out.println("msgQueue.size()=" + msgQueue.size());
        try {
            consume(poll);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void consume(Integer i) throws InterruptedException {
        TimeUnit.SECONDS.sleep(1);
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println(simpleDateFormat.format(new Date()) + "-> "+Thread.currentThread().getName() + " val:" + i);
    }
    public void consume2(Integer i, CountDownLatch latch) throws InterruptedException {
        TimeUnit.SECONDS.sleep(1);
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println(simpleDateFormat.format(new Date()) + "-> "+Thread.currentThread().getName() + " val:" + i);
        latch.countDown();
    }

    /**
     * 单个线程消费消息30条,花费30秒
     */
    public static void testSingleThread(){
        SingleConsumer singleConsumer = new SingleConsumer();
        singleConsumer.init(30);
        // 单个线程消费消息30条,花费30秒
        new Thread(()->{
            long start = System.currentTimeMillis();
            while (msgQueue.size() != 0) {
                try {
                    Integer take = msgQueue.poll();
                    singleConsumer.consume(take);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            long end = System.currentTimeMillis();
            System.out.println("cost "+(end-start));
        },"single").start();
    }

    /**
     * 30条数据 5个线程需要6秒,10个线程需要3秒,30线程需要1秒
     * 300条数据,100个线程需要3175ms,300个线程只要1秒
     *
     */
    public static void testMultiThread(){
        int initDataNum = 300;
        int threadNum = 300;
        SingleConsumer singleConsumer = new SingleConsumer();
        singleConsumer.init(initDataNum);
        CountDownLatch latch = new CountDownLatch(initDataNum);
        ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
        long start = System.currentTimeMillis();
        while (msgQueue.size() != 0) {
            Integer poll = msgQueue.poll();
            executorService.submit(() -> {
                try {
                    singleConsumer.consume2(poll,latch);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        try {
            latch.await();
            long end = System.currentTimeMillis();
            System.out.println("cost "+(end-start));
            executorService.shutdown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    public static void main(String[] args){

//        testSingleThread();
        testMultiThread();
    }


}
